package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2Partition {
  def main(args: Array[String]): Unit = {
    /**
      * 创建上下文对象
      *
      */

    val conf: SparkConf = new SparkConf()
      .setAppName("partition")
      .setMaster("local")


    //设置spark 默认分区数, 默认reduce端task的数量
    conf.set("spark.default.parallelism", "3")

    val sc = new SparkContext(conf)

    /**
      *
      * 1、读取文件的时候控制分区数
      * linesRDD分区数据默认等于block的数量
      * minPartitions： 由于控制读取文件得到的rdd的分区数
      * 分区数越多：task越多，任务越快，但是不能太多，因为task启动也需要时间
      * 保证每一个分区处理的数据量在128M左右
      * minPartitions : 只能用于增加分区不能用于减少分区
      */

    val linesRDD: RDD[String] = sc.textFile("data/words", 4)

    println(s"linesRDD分区数：${linesRDD.getNumPartitions}")

    /**
      * 窄依赖关系： 后免rdd的分区数等于前一个rdd的分区数据
      * 分区对应关系是一对一的关系
      *
      */

    val wordsRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))

    println(s"wordsRDD分区数：${wordsRDD.getNumPartitions}")

    /**
      * 宽依赖的算子产生的rdd的分区数默认等于前一个rdd的分区数
      * 所有会产生shuffle的算子都可以手动指定分区数据
      * 如果分区之后数据量减少了，可以减少分区，保证每一个分区的数据量在128M左右
      *
      * shuffle之后分区数优先级
      * 手动指定分区数 >  spark.default.parallelism  > 依赖上一个rdd
      *
      */

    val groupByRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy((word: String) => word, 10)


    println(s"groupByRDD分区数：${groupByRDD.getNumPartitions}")

    /*   groupByRDD.saveAsTextFile("data/temp")


       while (true) {

       }*/


  }

}
